import os
from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import *
from pyspark.sql.types import *

os.environ.setdefault('HADOOP_HOME', 'D:\\hadoop-2.9.2')

# Driver
spark = SparkSession \
    .builder \
    .master('local') \
    .appName('HelloSpark') \
    .getOrCreate()

df = spark.read.json('dataset/business.json')
df.show(5)
# 一、商户分析
# 1.找出美国最常见商户（前20)
# df.groupby('name') \
#     .count() \
#     .orderBy(col('count').desc()) \
#     .show(20, truncate=False)

# 2.找出美国商户最多的前10个城市
# df.groupby('city') \
#     .count() \
#     .orderBy(col('count').desc()) \
#     .show(10, truncate=False)

# 3.找出美国商户最多的前5个州
# df.groupby('state') \
#     .count() \
#     .orderBy(col('count').desc()) \
#     .show(5, truncate=False)

# 4.找出美国最常见商户，并显示平均评分（前20）
# df.groupby('name') \
#     .agg(avg('stars'), count('name').alias('count'))\
#     .orderBy(col('count').desc()) \
#     .show(20, truncate=False)

# 5.统计评分最高的城市（前10）
# df.groupby('city') \
#     .agg(max('stars'), count('city').alias('count'))\
#     .orderBy(col('count').desc()) \
#     .show(10, truncate=False)

# 6.统计category的数量
# df.withColumn('category', explode(split(col('categories'),',')))\
#     .agg(countDistinct('category'))\
#     .show()

# 7.统计最多的category及数量（前10）
# df.withColumn('category', explode(split(col('categories'),',')))\
#     .groupby('category')\
#     .count()\
#     .orderBy(col('count').desc()) \
#     .show(10, truncate=False)

# 8.收获五星评论最多的商户（前20）
# df.groupby('name', 'stars') \
#     .count()\
#     .orderBy(col('count').desc()) \
#     .where("stars == '5.0'") \
#     .show(20, truncate=False)

# 9.统计不同类型（中国菜、美式、墨西哥）的餐厅类型及数量
clean_df = df.filter(col('categories').like('%Restaurant%'))
# clean_df.withColumn('category', explode(split(col('categories'),',')))\
#     .groupby('category') \
#     .count()\
#     .orderBy(col('count').desc())\
#     .where("category like '%Chinese%' or category like '%American%' or category like '%Mexican%'")\
#     .show()

# 10.统计不同类型（中国菜、美式、墨西哥）的餐厅的评论数量
# clean_df.withColumn('category', explode(split(col('categories'),',')))\
#     .groupby('category') \
#     .agg(count('review_count').alias('review_count'))\
#     .orderBy(col('review_count').desc())\
#     .where("category like '%Chinese%' or category like '%American%' or category like '%Mexican%'")\
#     .show()

# 11.统计不同类型（中国菜、美式、墨西哥）的餐厅的评分分布
clean_df.withColumn('category', explode(split(col('categories'),',')))\
    .groupby('category', 'stars') \
    .count()\
    .orderBy(col('count').desc())\
    .where("category like '%Chinese%' or category like '%American%' or category like '%Mexican%'")\
    .show()